import java.awt.geom.Point2D

import com.zyl.common.SparkTools
import com.zyl.common.grid.Grid
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.functions._

object DwsStaypointMskD extends SparkTools {
  override def run(spark: SparkSession): Unit = {


    // 经度
    val getLongi: UserDefinedFunction = udf((grid: String) => {
      val point: Point2D.Double = Grid.getCenter(grid.toLong)
      point.getX
    })

    // 纬度
    val getLati: UserDefinedFunction = udf((grid: String) => {
      val point: Point2D.Double = Grid.getCenter(grid.toLong)
      point.getY
    })


    import spark.implicits._
    import org.apache.spark.sql.functions._

    spark
      //读取融合表
      .table("dwd.dwd_res_regn_mergelocation_msk_d")
      //取出指定分区
      .where($"day_id" === day_id)
      //获取开始时间和结束时间
      .withColumn("start_date", split($"start_time", ",")(1))
      .withColumn("end_date", split($"start_time", ",")(0))
      //按照手机号分组，按照时间排序，取上一条数据的网格编号
      .withColumn("s_grid", lag($"grid_id", 1, "") over Window.partitionBy($"mdn").orderBy($"start_date"))
      //判断当前网格编号和上一条数据的网格编号是否一致，划分边界
      .withColumn("flag", when($"s_grid" === $"grid_id", 0).otherwise(1))
      //在同一个组的数据打上同样的标记
      .withColumn("class", sum($"flag") over Window.partitionBy($"mdn").orderBy($"start_date"))
      //将同一个人在同一个网格中的数据分到同一个组
      .groupBy($"class", $"mdn", $"county_id", $"grid_id")
      //获取用户在网格中第一个点的时间和最后一个点的时间

      .agg(min($"start_date") as "grid_first_time", max($"end_date") as "grid_last_time")
      //获取停留时间
      .withColumn("stay_time", unix_timestamp($"grid_last_time", "yyyyMMddHHmmss") - unix_timestamp($"grid_first_time", "yyyyMMddHHmmss"))

      //使用udf函数获取网格中心点的经纬度
      .withColumn("longi", getLongi($"grid_id"))
      .withColumn("lati", getLati($"grid_id"))

      //整理数据
      .select($"mdn", $"longi", $"lati", $"grid_id", $"county_id", round($"stay_time" / 60, 4) as "stay_time", $"grid_first_time", $"grid_last_time")
      .write
      .mode(SaveMode.Overwrite)
      .format("csv")
      .option("sep", "\t")
      .save("/daas/motl/dws/dws_staypoint_msk_d/day_id=" + day_id)
  }
}
